package com.mo.redis;

import io.lettuce.core.RedisException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * @author moon.pan
 * @program mo-all
 * @apiNote
 * @since 2022/6/20
 */
@Slf4j
@Configuration
@ConditionalOnProperty(value = "spring.redis.host")
public class RedisStreamConfig {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private ListenerMessage listenerMessage;

    @Bean
    public StreamMessageListenerContainer<String, ObjectRecord<String, Object>> listenerContainer(RedisConnectionFactory factory) {
        // 创建Stream消息监听容器配置
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, Object>> options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                // 设置阻塞时间
                .pollTimeout(Duration.ofSeconds(1))
                // 配置消息类型
                .targetType(Object.class)
                .build();


        // 创建Stream消息监听容器
        StreamMessageListenerContainer<String, ObjectRecord<String, Object>> listenerContainer = StreamMessageListenerContainer
                .create(factory, options);
        // 监听容器启动
        listenerContainer.start();
        return listenerContainer;
    }

    @Bean
    public Subscription subscription(StreamMessageListenerContainer<String, ObjectRecord<String, Object>> listenerContainer) {
        checkGroup();
        // 设置消费手动提交配置
        return listenerContainer
                .register(StreamMessageListenerContainer.StreamReadRequest
                                // 设置订阅Stream的key和获取偏移量，以及消费处理类
                                .builder(StreamOffset.create("async_queue", ReadOffset.lastConsumed()))
                                // 设置消费者分组和名称
                                .consumer(Consumer.from("group1", "name1"))
                                // 手动ack
                                .autoAcknowledge(false)
                                .cancelOnError(throwable -> {
                                    throwable.printStackTrace();
                                    return false;
                                })
                                .build(),
                        listenerMessage);
    }

//    @Bean
//    public Subscription subscription2(StreamMessageListenerContainer<String, ObjectRecord<String, String>> listenerContainer) {
//        return listenerContainer.receiveAutoAck(
//                Consumer.from("group1", "name2"),
//                StreamOffset.create("key", ReadOffset.lastConsumed()),
//                streamListener
//        );
//    }
//
//    @Bean
//    public Subscription subscription3(StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamMessageListenerContainer) {
//        return streamMessageListenerContainer.receive(
//                Consumer.from("group2", "name1"),
//                StreamOffset.create("stream1", ReadOffset.lastConsumed()),
//                streamListener
//        );
//    }

    /**
     * 由于订阅需要先有stream，先做下检查
     */
    private void checkGroup() {
        // 创建需要校验的分组List
        List<String> consumers = new ArrayList<>();
        consumers.add("group1");
        StreamInfo.XInfoGroups infoGroups = null;
        try {
            // 获取Stream的所有组信息
            infoGroups = redisTemplate.opsForStream().groups("async_queue");
        } catch (RedisSystemException | RedisException | InvalidDataAccessApiUsageException ex) {
//            log.error("group key not exist or commend error", ex);
        }

        // 遍历校验分组是否存在
        for (String consumer : consumers) {
            boolean consumerExist = false;
            if (Objects.nonNull(infoGroups)) {
                if (infoGroups.stream().anyMatch(t -> Objects.equals(consumer, t.groupName()))) {
                    consumerExist = true;
                }
            }
            // 创建不存在的分组
            if (!consumerExist) {
                redisTemplate.opsForStream().createGroup("async_queue", consumer);
            }
        }

    }
}
